package com.atguigu.flink.chapter05.transform;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * @Author lzc
 * @Date 2023/6/19 13:47
 */
public class SplitDemo2 {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        DataStreamSource<Integer> stream1 = env.fromElements(1, 2, 3, 4, 5, 6, 9);
        // 主流: 奇数
        // 测输出流: 偶数
        SingleOutputStreamOperator<Integer> oddStream = stream1
            .process(new ProcessFunction<Integer, Integer>() {
                // 流中每来一个元素, 这个方法就执行一次
                // 参数 1: 需要处理的元素
                // 参数 2: 上下文对象, 环境
                // 参数 3: 收集器
                @Override
                public void processElement(Integer value,
                                           Context ctx,
                                           Collector<Integer> out) throws Exception {
                    if (value % 2 == 1) {
                        out.collect(value);
                    } else {
                        // 放入测输出流
                        // 向测输出流输出数据
                        ctx.output(new OutputTag<Integer>("even"){}, value);
                    }
                    
                    if(value > 3){
                        ctx.output(new OutputTag<Integer>("greater_three"){}, value);
                    }
                }
            });
    
        SideOutputDataStream<Integer> evenStream = oddStream.getSideOutput(new OutputTag<Integer>("even"){});
    
        oddStream.print("odd");
        evenStream.print("even");
        oddStream.getSideOutput(new OutputTag<Integer>("greater_three") {}).print("greater_three");
    
    
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
